package cn.doitedu.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job04_订单日清日结看板 {

    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        TableConfig config = tenv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay", "1 s");



        // 创建cdc连接器表，映射业务库中的  oms_order 表
        tenv.executeSql(
                " create table oms_order_mysql (        "+
                        "     id bigint                                    "+
                        "     ,member_id bigint                            "+
                        "     ,create_time timestamp(3)                    "+
                        "     ,total_amount decimal(10,2)                  "+
                        "     ,pay_amount decimal(10,2)                    "+
                        "     ,promotion_amount decimal(10,2)              "+
                        "     ,integration_amount decimal(10,2)            "+
                        "     ,coupon_amount decimal(10,2)                 "+
                        "     ,status  int                                 "+
                        "     ,payment_time  timestamp(3)                  "+
                        "     ,delivery_time  timestamp(3)                 "+
                        "     ,receive_time  timestamp(3)                  "+
                        "     ,modify_time  timestamp(3)                   "+
                        "     ,rt as modify_time                           "+
                        "     ,watermark for rt as rt                      "+
                        "     ,primary key(id) not enforced     "+
                        " ) WITH (                              "+
                        "     'connector' = 'mysql-cdc',        "+
                        "     'hostname' = 'doitedu',           "+
                        "     'port' = '3306',                  "+
                        "     'username' = 'root',              "+
                        "     'password' = 'root',              "+
                        "     'database-name' = 'realtimedw',   "+
                        "     'table-name' = 'oms_order'      "+
                        " )                                     "
        );

        // 统计sql
/*        tenv.executeSql(
                " SELECT                                                                                                                                 "+
                        "    window_start                                                                                                               "+
                        " 	,window_end                                                                                                                  "+
                        " 	,count(1) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE)                  as `订单总数`                   "+
                        " 	,sum(total_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE)         as `订单gmv总额`                "+
                        " 	,sum(pay_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE)           as `订单应付总额`               "+
                        " 	,sum(pay_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE)           as `订单实付总额`               "+
                        " 	,count(1) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 0 )  as `待支付订单数`               "+
                        " 	,sum(total_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 0 )  `待支付订单原价总额`    "+
                        " 	,count(1) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 1 )           `已支付订单数`         "+
                        " 	,sum(total_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 1 )  `已支付订单原价总额`    "+
                        " 	,count(1) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 2 )           `已发货订单数`         "+
                        " 	,sum(total_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 2 )  `已发货订单原价总额`    "+
                        " 	,count(1) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 3 )           `已确认订单数`         "+
                        " 	,sum(total_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 3 )  `已确认订单原价总额`    "+
                        " from table(                                                                                                                    "+
                        "     cumulate(table oms_order_mysql ,descriptor(rt),interval '1' second,interval '24' hour)                                     "+
                        " )                                                                                                                              "+
                        " group by                                                                                                                       "+
                        "     window_start,                                                                                                              "+
                        " 	  window_end                                                                                                                 "
        ).print();*/

        // groupBy window聚合，支持消费changelog流
        // 但是：groupBy window不存在cumulate窗口，只有 tumble 和  hop
        // 变通： 利用tumble + flinkSql底层提供的一个“提前触发器”，来模拟实现cumulate的效果
        tenv.executeSql(
                " SELECT                                                                                                                                    "+
                        "    tumble_start(rt,interval '24' hour)   as window_start                                                                          "+
                        " 	,tumble_end(rt,interval '24' hour)    as window_end                                                                             "+
                        " 	,count(1) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE)                  as `订单总数`                      "+
                        " 	,sum(total_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE)         as `订单gmv总额`                   "+
                        " 	,sum(pay_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE)           as `订单应付总额`                  "+
                        " 	,sum(pay_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE)           as `订单实付总额`                  "+
                        " 	,count(1) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 0 )  as `待支付订单数`                  "+
                        " 	,sum(total_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 0 )  `待支付订单原价总额`      "+
                        " 	,count(1) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 1 )           `已支付订单数`            "+
                        " 	,sum(total_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 1 )  `已支付订单原价总额`      "+
                        " 	,count(1) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 2 )           `已发货订单数`            "+
                        " 	,sum(total_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 2 )  `已发货订单原价总额`      "+
                        " 	,count(1) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 3 )           `已确认订单数`            "+
                        " 	,sum(total_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE AND status = 3 )  `已确认订单原价总额`      "+
                        " from  oms_order_mysql                                                                                                            "+
                        " group by                                                                                                                         "+
                        "      tumble(rt,interval '24' hour)                                                                          "
        ).print();




    }
}
